package com.xqboss.common.core.mq;

import cn.hutool.core.date.LocalDateTimeUtil;
import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.serializer.SerializerFeature;
import com.xqboss.common.config.properties.RocketMqConfig;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.apis.ClientServiceProvider;
import org.apache.rocketmq.client.apis.message.Message;
import org.apache.rocketmq.client.apis.producer.Producer;
import org.apache.rocketmq.client.apis.producer.SendReceipt;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Lazy;
import org.springframework.stereotype.Component;

import javax.annotation.PreDestroy;
import java.io.IOException;
import java.nio.charset.StandardCharsets;
import java.time.LocalDateTime;


/**
 * @Author jjj
 * mq消息发送唯一入口
 */
@Slf4j
@Component
public class MqSender {

    @Lazy
    @Autowired
    protected Producer producer;

    @Autowired
    private RocketMqConfig rocketMqConfig;

//    /**
//     * 埋点公用mq生产者
//     *
//     * @param buryModel
//     */
//    public boolean record(BuryModel buryModel, RecordMqEnum recordMqEnum) {
//        try {
//            log.info("埋点消息生产,recordMqEnum:{},message:{}", recordMqEnum, JSON.toJSONString(buryModel));
//            Message message = ClientServiceProvider.loadService().newMessageBuilder()
//                    .setTopic(recordMqEnum.getTopic())
//                    .setTag(recordMqEnum.getTag())
//                    .setBody(JSON.toJSONString(buryModel, SerializerFeature.WriteMapNullValue).getBytes(StandardCharsets.UTF_8))
//                    .addProperty(SleuthDto.MQ_KEY, traceUtil.sleuth())
//                    .build();
//            // 异步发送，不处理结果
//            SendReceipt send = producer.send(message);
//            log.info("埋点消息生产结果,messageId:{},recordMqEnum:{},message:{}", send.getMessageId(), recordMqEnum, JSON.toJSONString(buryModel));
//            return true;
//        } catch (Exception e) {
//            log.error("mq统计消息发送失败,enum:{}", recordMqEnum, e);
//            return false;
//        }
//    }

    /**
     * 业务mq发送者
     *
     * @param object
     */
    public boolean send(MqNormalEnum mqEnum, Object object) {
        try {
            String business = mqEnum.getMessage();
            log.info("业务mq发送者生产, business： {}, message:{}", business, JSON.toJSONString(object));
            Message message = ClientServiceProvider.loadService().newMessageBuilder()
                    .setTopic(rocketMqConfig.getTopicPrefix() + mqEnum.getTopic())
                    .setTag(mqEnum.getTag())
                    .setBody(JSON.toJSONString(object, SerializerFeature.WriteMapNullValue).getBytes(StandardCharsets.UTF_8))
//                    .addProperty(SleuthDto.MQ_KEY, traceUtil.sleuth())
                    .build();
            SendReceipt sendResult = producer.send(message);
            log.info("发送结果, business： {} topic:{},tag:{},messageId:{}", business,
                    mqEnum.getTopic(), mqEnum.getTag(), sendResult.getMessageId());
            return true;
        } catch (Exception e) {
            log.error("业务mq发送消息失败,enum:{}", mqEnum, e);
            return false;
        }
    }

    /**
     * afterTime 秒，发送后多少秒后消费
     *
     * @param object
     * @param mqEnum
     * @return
     */
    public boolean sendDelay(MqDelayEnum mqEnum, Object object, Long afterTime) {
        return sendDelay(mqEnum, object,LocalDateTime.now().plusSeconds(afterTime));
    }

    /**
     * afterTime 秒，发送后多少秒后消费
     *
     * @param object
     * @param mqEnum
     * @return
     */
    public boolean sendDelay(MqDelayEnum mqEnum, Object object, LocalDateTime deliveryTime) {
        String business = mqEnum.getMessage();
        log.info("业务mq发送者生产, business：{},message:{}", business, JSON.toJSONString(object));
        try {
            // 延时消息需要测试
            Message message = ClientServiceProvider.loadService().newMessageBuilder()
                    .setTopic(rocketMqConfig.getTopicPrefix() + mqEnum.getTopic())
                    .setTag(mqEnum.getTag())
                    .setBody(JSON.toJSONString(object, SerializerFeature.WriteMapNullValue).getBytes(StandardCharsets.UTF_8))
                    .setDeliveryTimestamp(LocalDateTimeUtil.toEpochMilli(deliveryTime))
//                    .addProperty(SleuthDto.MQ_KEY, traceUtil.sleuth())
                    .build();
            SendReceipt sendResult = producer.send(message);
            log.info("发送结果, business：{},topic:{},tag:{},messageId:{}", business,
                    mqEnum.getTopic(), mqEnum.getTag(), sendResult.getMessageId());
            return true;
        } catch (Exception e) {
            log.error("发送结果失败{},enum:{}", mqEnum, e);
            return false;
        }
    }

    /**
     * 取消延迟消息，暂不支持
     */
    public void cancelDelay(MqDelayEnum mqEnum, Object object) {

    }

}
